package com.vertx.kafka;

import io.vertx.core.Vertx;
import io.vertx.kafka.client.consumer.KafkaConsumer;
import io.vertx.kafka.client.consumer.KafkaConsumerRecord;

import java.time.Duration;
import java.util.HashMap;

public class KafkaConsumerCommit {

    public static void main(String[] args) {

        var config = new HashMap<String, String>();

        config.put("bootstrap.servers", "172.18.30.14:9092");
        config.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        config.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        config.put("group.id", "zhengxing-sync-test.zhengxing.statistic_device-01");
        config.put("auto.offset.reset", "earliest");
        config.put("enable.auto.commit", "true");

        var vertx = Vertx.vertx();
        KafkaConsumer<String, String> consumer = KafkaConsumer.create(vertx, config);

        var topic = "zhengxing-sync-test.zhengxing.statistic_device";

        // 订阅topic
        consumer.subscribe(topic)
                .onSuccess(h->{
                    System.out.println("subscribe success");
                })
                .onFailure(e->{
                    System.out.println("subscribe error " + e);
                });
        // 分区处理
        consumer.partitionsAssignedHandler(h->{
            System.out.println("partition assigned");
            h.forEach(tp->{
                System.out.println(tp.getPartition());
            });
        })
                .partitionsRevokedHandler(h->{
                    System.out.println("partition revoked");
                    h.forEach(tp->{
                        System.out.println(tp.getPartition());
                    });
                });

        vertx.setPeriodic(2000, timerId -> {
            System.out.println("*******************");
            consumer
                    // pool(timeout)  阻塞获取数据的时间
                    .poll(Duration.ofMillis(10))
                    .onSuccess(records -> {
                        System.out.println("-------------------------------" + records.size());
                        for (int i = 0; i < records.size(); i++) {
                            KafkaConsumerRecord<String, String> record = records.recordAt(i);
                            System.out.println("key=" + record.key() + ",value=" + record.value() +
                                    ",partition=" + record.partition() + ",offset=" + record.offset());
                            // 提交订阅的topic的当前offset
                            consumer.commit()
                                    .onSuccess(h->{
                                        System.out.println("commit success");
                                    });
                        }
                    })
                    .onFailure(cause -> {
                        System.out.println("Something went wrong when polling " + cause.toString());
                        cause.printStackTrace();

                        // Stop polling if something went wrong
                        vertx.cancelTimer(timerId);
                    });
        });
    }

}
